/*
 * ===========================================================================
 *
 *      Filename: work_queue.c
 *   Description: implement for work queue
 *        Author: Ye wenhao, wenhaoye@126.com
 *  Organization: 
 *
 * ===========================================================================
 */
#include "list.h"
#include "work_queue.h"
#include <sys/eventfd.h>

typedef struct wqueue {
    work_t      *head;
    work_t      *tail;
    uint32_t    nums;
	mutex_t		lock;
}wqueue_t;

struct worker {
    wqueue_t        wq;
    int             efd;
    task_t          task;
    uint8_t         exit:1;
    uint8_t         reverse:7;
};


/*****************************************************************************
 *                         work queue
 ****************************************************************************/
static int wqueue_init(wqueue_t *self)
{
    self->head = NULL;
    self->tail = NULL;
    self->nums = 0;
	mutex_init(&self->lock);
    return 0;
}

static void wqueue_term(wqueue_t *self)
{
	mutex_term(&self->lock);
    self->nums = 0;
    self->head = NULL;
    self->tail = NULL;
}

static void wqueue_destroy(wqueue_t **p_self)
{
    if (p_self && *p_self) {
        wqueue_t *self = *p_self;
        wqueue_term(self);
        safe_free(*p_self);
    }
}

static wqueue_t *wqueue_new(void)
{
    wqueue_t *self = (wqueue_t *)MALLOC(sizeof(wqueue_t));
    if (self) {
        if (wqueue_init(self) < 0)
            wqueue_destroy(&self);
    }

    return self;
}

static inline void wqueue_lock(wqueue_t *self)
{
    return mutex_get(&self->lock);
}

static inline void wqueue_unlock(wqueue_t *self)
{
    return mutex_put(&self->lock);
}

static void wqueue_add(wqueue_t *self, work_t *w)
{
    wqueue_lock(self);
    /* add to head firstly */
    w->next = self->head;
    self->head = w;

    if (!self->tail)
        self->tail = w;

    self->nums++;
    wqueue_unlock(self);
}

static void wqueue_add_tail(wqueue_t *self, work_t *w)
{
    w->next = NULL;

    wqueue_lock(self);
    /* add to tail firstly */
    if (self->tail)
        self->tail->next = w;
    self->tail = w;

    if (!self->head)
        self->head = w;
    
    self->nums++;
    wqueue_unlock(self);
}

static work_t *wqueue_get_head(wqueue_t *self)
{
    work_t *head = NULL;
    
    wqueue_lock(self);
    head = self->head;
    self->head = NULL;
    self->tail = NULL;
    //self->nums = 0;
    wqueue_unlock(self);
    
    return head;
}

static work_t *wqueue_get(wqueue_t *self)
{
    work_t *work = NULL;
    
    wqueue_lock(self);
    work = self->head;
    self->head = work->next;

    if (self->tail == work)
        self->tail = NULL;
    
    self->nums--;
    wqueue_unlock(self);

    work->next = NULL;
    return work;
}

static uint32_t wqueue_nums(wqueue_t *self)
{
	uint32_t nums = 0;
    wqueue_lock(self);
	nums = self->nums;
    wqueue_unlock(self);

	return nums;
}

static uint32_t wqueue_nums_dec(wqueue_t *self)
{
	uint32_t nums = 0;
    wqueue_lock(self);
	nums = --self->nums;
    wqueue_unlock(self);

	return nums;
}

/*****************************************************************************
 *                           worker
 ****************************************************************************/
static void _worker_term(worker_t *self)
{
    self->exit = 1;
    safe_close(self->efd);
    wqueue_term(&self->wq);
}

static void *worker_thr(void *arg)
{
    worker_t *self = (worker_t *)arg;
    uint64_t count;
    work_t *cur;
    work_t *next;

    pthread_detach(pthread_self());
    while (!self->exit) {
        if (read(self->efd, &count, sizeof(count)) < 0) {
            ERR("read fail");
            self->exit = 1;
            break;
        }

        cur = wqueue_get_head(&self->wq);
        //cur = wqueue_get(&self->wq);
        while (!self->exit && cur) {
            next = cur->next;
            cur->next = NULL;
            if (cur->func)
                cur->func(cur->arg);
            cur = next;

			wqueue_nums_dec(&self->wq);
        }
    }

    _worker_term(self);
    return NULL;
}

int worker_init(worker_t *self)
{
    self->exit = 0;
    
    if (wqueue_init(&self->wq) < 0)
        return -1;
    
    self->efd = eventfd(0, EFD_CLOEXEC);
    if (self->efd < 0) {
        ERR("eventfd fail");
        goto _fail;
    }

    if (task_init(&self->task, "work-thr", worker_thr, (void *)self) < 0)
        goto _fail;

    return 0;

_fail:
    _worker_term(self);
    return -1;
}

void worker_term(worker_t *self)
{
    self->exit = 1;
    if (self->efd > 0) {
        uint64_t count = 1;
        write(self->efd, &count, sizeof(count));
    }
}

worker_t *worker_new(void)
{
    worker_t *self = (worker_t *)MALLOC(sizeof(worker_t));
    if (self) {
        if (worker_init(self) < 0)
            worker_destroy(&self);
    }

    return self;
}

void worker_destroy(worker_t **p_self)
{
    if (p_self && *p_self) {
        worker_t *self = *p_self;
        _worker_term(self);
        safe_free(*p_self);
    }
}

void worker_add(worker_t *self, work_t *w)
{
    wqueue_add(&self->wq, w);
    uint64_t count = 1;
    write(self->efd, &count, sizeof(count));
}

void worker_add_tail(worker_t *self, work_t *w)
{
    wqueue_add_tail(&self->wq, w);
    uint64_t count = 1;
    write(self->efd, &count, sizeof(count));
}

uint32_t worker_nums(worker_t *self)
{
    return wqueue_nums(&self->wq);
}
